package com.edais.rabbit;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.edais.service.SMSService;
import com.edais.util.Env;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;

@Service
public class RabbitUtil {
	private static final Log logger = LogFactory.getLog(RabbitUtil.class);
	
	@Autowired
	RabbitTemplate rabbitTemplate;
	
	@Autowired
	private SMSService sMSService;

	private RabbitUtil() {
	}

	private static final String host = Env.getValue("rabbit.host");
	private static final int port = Integer.valueOf(Env.getValue("rabbit.port"));
	
	private static final String username = Env.getValue("rabbit.username");
	private static final String password = Env.getValue("rabbit.password");

	public static final String QUEUE = "quene";
	public static final String DELAY_1S_QUENE = "delay_1s_quene";
	public static final String DELAY_1M_QUENE = "delay_1m_quene";
	public static final String DELAY_5M_QUENE = "delay_5m_quene";
	public static final String DELAY_15M_QUENE = "delay_15m_quene";
	public static final String DELAY_30M_QUENE = "delay_30m_quene";
	public static final String DELAY_120M_QUENE = "delay_120m_quene";

	/**
	 * @param logInfo   日志头
	 * @param queneName 队列名称
	 * @param message   消息
	 */
	public void convertAndSend(String logInfo, String queneName, String message, String phoneNumber, String reqOrderId) {
		logger.info(logInfo + "发送rabbit报文开始" + host + ":" + port + "-" + queneName + "-" + message);
		try {
			rabbitTemplate.convertAndSend(queneName, message.getBytes("utf-8"));
		} catch (Exception e) {
			logger.error(logInfo + "异常", e);
			sMSService.sendErrorMessageForSystem("sms05004","",phoneNumber, reqOrderId, "发送rabbitmq失败，请联系运维重启");
		}
		logger.info(logInfo + "发送rabbit报文结束");
	}
	
	/**
	 * @param logInfo   日志头
	 * @param queneName 队列名称
	 * @param message   消息
	 */
	public static void sendMessage(String logInfo, String queneName, String message) {
		logger.info(logInfo + "发送rabbit报文开始" + host + ":" + port + "-" + queneName + "-" + message);
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost(host);
		factory.setPort(port);
		factory.setUsername(username);
		factory.setPassword(password);
		Connection connection = null;
		Channel channel = null;
		try {
			connection = factory.newConnection();
			channel = connection.createChannel();

			String exchange = ""; // the exchange to publish the message to
			String routingKey = queneName; // the routing key
			// props other properties for the message - routing headers etc
			BasicProperties props = new AMQP.BasicProperties.Builder().deliveryMode(2).build(); // 2-Persistent
			byte[] body = message.getBytes("utf-8"); // the message body
			channel.basicPublish(exchange, routingKey, props, body);
		} catch (IOException e) {
			logger.error(logInfo + "异常", e);
		} catch (TimeoutException e) {
			logger.error(logInfo + "异常", e);
		} catch (Exception e) {
			logger.error(logInfo + "异常", e);
		} finally {
			try {
				if (channel != null) {
					channel.close();
				}
			} catch (IOException e) {
				logger.error(logInfo + "异常", e);
			} catch (TimeoutException e) {
				logger.error(logInfo + "异常", e);
			} catch (Exception e) {
				logger.error(logInfo + "异常", e);
			}
			try {
				if (connection != null) {
					connection.close();
				}
			} catch (IOException e) {
				logger.error(logInfo + "异常", e);
			} catch (Exception e) {
				logger.error(logInfo + "异常", e);
			}
		}
		logger.info(logInfo + "发送rabbit报文结束");
	}
	
	public static void bind(final String queneName, final RabbitConsumer<String> rabbitConsumer){
		new Thread() {
			@Override
			public void run() {
				ConnectionFactory factory = new ConnectionFactory();
				factory.setHost(host);
				factory.setPort(port);
				factory.setUsername(username);
				factory.setPassword(password);
				Connection connection = null;
				Channel channel = null;

				try {
					connection = factory.newConnection();
					channel = connection.createChannel();
					QueueingConsumer consumer = new QueueingConsumer(channel);
					// 指定消费队列
					channel.basicConsume(queneName, true, consumer);
					while (true) {
						QueueingConsumer.Delivery delivery = consumer.nextDelivery();
						String message = new String(delivery.getBody(), "utf-8");
						logger.info("rabbitmq接收到数据" + message);
						if (rabbitConsumer == null) {
							logger.error("rabbitmq消费者为空" + message);
						} else {
							rabbitConsumer.accept(message);
						}
					}
				} catch (IOException e) {
					logger.error("异常", e);
				} catch (TimeoutException e) {
					logger.error("异常", e);
				} catch (Exception e) {
					logger.error("异常", e);
				} finally {
					try {
						if (channel != null) {
							channel.close();
						}
					} catch (IOException e) {
						logger.error("异常", e);
					} catch (TimeoutException e) {
						logger.error("异常", e);
					} catch (Exception e) {
						logger.error("异常", e);
					}
					try {
						if (connection != null) {
							connection.close();
						}
					} catch (IOException e) {
						logger.error("异常", e);
					} catch (Exception e) {
						logger.error("异常", e);
					}
				}
			}
		}.start();
	}
}